package com.bw.dws;

import com.bw.bean.DeviceData;
import com.bw.util.ProfileCalcUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import java.sql.Timestamp;
import java.util.HashMap;
import java.util.Map;

/**
 * 身高体重标签实时计算（多源优先级）
 * 文档参考：🔶1-193（设备权重1.0>订单0.8>会员0.6）、🔶1-200（同权重取最新）
 */
public class BodyTagCalc extends KeyedProcessFunction<String, DeviceData, Map<String, Object>> {
    // 状态存储：身高/体重的最新值+来源优先级+更新时间
    private ValueState<Map<String, Object>> heightState;
    private ValueState<Map<String, Object>> weightState;

    // 数据来源优先级：device=3 > order=2 > member=1（数值越大优先级越高）
    private int getSourcePriority(String sourceType) {
        switch (sourceType) {
            case "device": return 3;
            case "order": return 2;
            case "member": return 1;
            default: return 0;
        }
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        // 初始化身高状态（key：value=值，priority=优先级，updateTime=时间）
        heightState = getRuntimeContext().getState(new ValueStateDescriptor<>("heightState", 
                org.apache.flink.api.common.typeinfo.Types.MAP(
                    org.apache.flink.api.common.typeinfo.Types.STRING,
                    org.apache.flink.api.common.typeinfo.Types.GENERIC(Object.class)
                )));
        weightState = getRuntimeContext().getState(new ValueStateDescriptor<>("weightState", 
                org.apache.flink.api.common.typeinfo.Types.MAP(
                    org.apache.flink.api.common.typeinfo.Types.STRING,
                    org.apache.flink.api.common.typeinfo.Types.GENERIC(Object.class)
                )));
    }

    @Override
    public void processElement(DeviceData data, Context ctx, Collector<Map<String, Object>> out) throws Exception {
        String userId = data.getUserId();
        String dataType = data.getDataType();
        String sourceType = "device"; // 当前处理设备数据，订单/会员数据需扩展Source
        Timestamp syncTime = data.getSyncTime();

        // （1）数据标准化
        Double standardValue = null;
        if ("height".equals(dataType)) {
            standardValue = ProfileCalcUtil.standardizeHeight(data.getDataValue());
        } else if ("weight".equals(dataType)) {
            standardValue = ProfileCalcUtil.standardizeWeight(data.getDataValue());
        }
        if (standardValue == null) return; // 异常值过滤

        // （2）按优先级更新状态
        Map<String, Object> result = new HashMap<>();
        result.put("userId", userId);
        result.put("updateTime", new Timestamp(System.currentTimeMillis()));

        if ("height".equals(dataType)) {
            Map<String, Object> currentHeight = heightState.value() == null ? new HashMap<>() : heightState.value();
            int currentPriority = (int) currentHeight.getOrDefault("priority", 0);
            int newPriority = getSourcePriority(sourceType);

            // 优先级更高 或 同优先级时间更新 → 覆盖
            if (newPriority > currentPriority || 
                (newPriority == currentPriority && syncTime.after((Timestamp) currentHeight.getOrDefault("updateTime", new Timestamp(0))))) {
                currentHeight.put("value", standardValue);
                currentHeight.put("priority", newPriority);
                currentHeight.put("updateTime", syncTime);
                heightState.update(currentHeight);
            }
            result.put("height", currentHeight.get("value"));
        } else if ("weight".equals(dataType)) {
            Map<String, Object> currentWeight = weightState.value() == null ? new HashMap<>() : weightState.value();
            int currentPriority = (int) currentWeight.getOrDefault("priority", 0);
            int newPriority = getSourcePriority(sourceType);

            if (newPriority > currentPriority || 
                (newPriority == currentPriority && syncTime.after((Timestamp) currentWeight.getOrDefault("updateTime", new Timestamp(0))))) {
                currentWeight.put("value", standardValue);
                currentWeight.put("priority", newPriority);
                currentWeight.put("updateTime", syncTime);
                weightState.update(currentWeight);
            }
            result.put("weight", currentWeight.get("value"));
        }

        // （3）输出结果
        out.collect(result);
    }
}